package com.larry.rabbitmq.fourth;

import com.larry.rabbitmq.util.RabbitMQUtils;
import com.rabbitmq.client.Channel;

import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @author larry
 * @Title: larry-rabbitmq
 * @Package com.larry.rabbitmq.fourth
 * @Description: 发布确认
 * @date 6/29/212:07 PM
 */
public class ConfirmMessage {

    public static final int MESSAGE_COUNT = 1000;

    public static void main(String args[]) throws Exception{

        //单个确认    耗时: 424ms
//        ConfirmMessage.publishMessageIndividually();

        //批量确认    耗时: 62ms
//        ConfirmMessage.publishMessageBatch();

        //异步确认    耗时: 42ms
        ConfirmMessage.publishMessageAsync();
    }


    public static void publishMessageIndividually() throws Exception{
        Channel channel = RabbitMQUtils.getChannel();

        //声明队列
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,false,false,false,null);

        //开启发布确认
        channel.confirmSelect();

        //开始时间
        long begin = System.currentTimeMillis();

        //批量发消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("",queueName,null,message.getBytes());
            //单个消息进行确认
            boolean b = channel.waitForConfirms();
            if (b){
                System.out.println("消息发布成功....");
            }
        }

        //结束时间
        long end = System.currentTimeMillis();

        System.out.println("耗时: " + (end - begin) + "ms");
    }



    //批量发布确认
    public static void publishMessageBatch() throws Exception{
        Channel channel = RabbitMQUtils.getChannel();

        //声明队列
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,false,false,false,null);

        //开启发布确认
        channel.confirmSelect();

        //批量确认个数
        int batchSize = 100;

        //开始时间
        long begin = System.currentTimeMillis();

        //批量发消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("",queueName,null,message.getBytes());

            if (i % batchSize == 0){
                channel.waitForConfirms();
            }
        }

        //结束时间
        long end = System.currentTimeMillis();

        System.out.println("耗时: " + (end - begin) + "ms");
    }

    //异步发布确认
    public static void publishMessageAsync() throws Exception{
        Channel channel = RabbitMQUtils.getChannel();

        //声明队列
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,false,false,false,null);

        //开启发布确认
        channel.confirmSelect();

        /**
         * 线程安全有序的哈希表 适用于高并发情况下
         *  将序号与消息进行关联
         *  批量删除条目
         *  支持高并发 多线程
         */
        ConcurrentSkipListMap<Long,String> concurrentSkipListMap = new ConcurrentSkipListMap();

        //批量确认个数
        int batchSize = 100;

        //开始时间
        long begin = System.currentTimeMillis();

        //消息监听器   成功回调  失败回调  异步通知
        //消息标记  是否为批量确认
        channel.addConfirmListener((deliveryTag, multiple) -> {
            //删除确认消息
            if (multiple){
                ConcurrentNavigableMap<Long, String> map = concurrentSkipListMap.headMap(deliveryTag);
                map.clear();
            }else {
                concurrentSkipListMap.remove(deliveryTag);
            }
            System.out.println("确认的消息: " + deliveryTag);
        },(deliveryTag, multiple) -> {
            String message = concurrentSkipListMap.get(deliveryTag);
            System.out.println("未确认的消息: " + deliveryTag + "---" + message);
        });

        //批量发消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("",queueName,null,message.getBytes());

            //将消息记录
            concurrentSkipListMap.put(channel.getNextPublishSeqNo(),message);
        }

        //结束时间
        long end = System.currentTimeMillis();

        System.out.println("-------------------耗时: " + (end - begin) + "ms");
    }

}
